package com.interview.lock;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 高效消息队列
 * 
 * 核心设计思想：
 * 1. 使用ConcurrentLinkedQueue作为底层存储 - 无锁并发队列，支持高并发写入
 * 2. 使用ReentrantLock + Condition实现消费者的等待/唤醒机制
 * 3. 生产者无需获取锁，直接写入队列（高性能）
 * 4. 消费者使用锁来保证单线程消费和等待机制
 * 5. 支持批量消费以提高吞吐量
 */
public class MessageQueue {
    
    // 使用ConcurrentLinkedQueue作为底层队列
    // 优点：无锁设计，支持高并发的生产者写入，性能极高
    private final ConcurrentLinkedQueue<Message> queue;
    
    // 用于消费者等待/唤醒的锁和条件变量
    // 只有消费者会使用这个锁，生产者不需要
    private final ReentrantLock consumerLock;
    private final Condition notEmpty;
    
    // 统计信息
    private final AtomicLong totalProduced;  // 总生产消息数
    private final AtomicLong totalConsumed;  // 总消费消息数
    
    // 队列容量限制（可选，防止内存溢出）
    private final int capacity;
    
    public MessageQueue() {
        this(Integer.MAX_VALUE);
    }
    
    public MessageQueue(int capacity) {
        this.queue = new ConcurrentLinkedQueue<>();
        this.consumerLock = new ReentrantLock();
        this.notEmpty = consumerLock.newCondition();
        this.totalProduced = new AtomicLong(0);
        this.totalConsumed = new AtomicLong(0);
        this.capacity = capacity;
    }
    
    /**
     * 生产者添加消息
     * 
     * 关键设计：
     * - 不使用锁，直接调用ConcurrentLinkedQueue的offer方法
     * - ConcurrentLinkedQueue内部使用CAS操作，保证线程安全
     * - 添加成功后，唤醒可能在等待的消费者
     * 
     * @param message 要添加的消息
     * @return 是否添加成功
     */
    public boolean offer(Message message) {
        // 检查容量限制
        if (totalProduced.get() - totalConsumed.get() >= capacity) {
            return false;
        }
        
        // 直接添加到队列，无需加锁
        boolean success = queue.offer(message);
        
        if (success) {
            totalProduced.incrementAndGet();
            
            // 通知消费者有新消息
            // 这里需要获取锁来使用Condition
            consumerLock.lock();
            try {
                notEmpty.signal();
            } finally {
                consumerLock.unlock();
            }
        }
        
        return success;
    }
    
    /**
     * 消费者获取单个消息（阻塞）
     * 
     * @return 消息对象
     * @throws InterruptedException
     */
    public Message take() throws InterruptedException {
        consumerLock.lock();
        try {
            // 如果队列为空，等待
            while (queue.isEmpty()) {
                notEmpty.await();
            }
            
            Message message = queue.poll();
            if (message != null) {
                totalConsumed.incrementAndGet();
            }
            return message;
        } finally {
            consumerLock.unlock();
        }
    }
    
    /**
     * 消费者批量获取消息（阻塞）
     * 
     * 批量获取可以显著提高消费效率，减少锁竞争
     * 
     * @param maxSize 最大获取数量
     * @return 消息列表
     * @throws InterruptedException
     */
    public List<Message> takeBatch(int maxSize) throws InterruptedException {
        consumerLock.lock();
        try {
            // 如果队列为空，等待
            while (queue.isEmpty()) {
                notEmpty.await();
            }
            
            List<Message> messages = new ArrayList<>(maxSize);
            
            // 批量获取消息
            for (int i = 0; i < maxSize; i++) {
                Message message = queue.poll();
                if (message == null) {
                    break;
                }
                messages.add(message);
                totalConsumed.incrementAndGet();
            }
            
            return messages;
        } finally {
            consumerLock.unlock();
        }
    }
    
    /**
     * 非阻塞方式获取消息
     * 
     * @return 消息对象，如果队列为空返回null
     */
    public Message poll() {
        Message message = queue.poll();
        if (message != null) {
            totalConsumed.incrementAndGet();
        }
        return message;
    }
    
    /**
     * 获取队列当前大小
     * 
     * @return 队列大小
     */
    public int size() {
        return queue.size();
    }
    
    /**
     * 判断队列是否为空
     * 
     * @return 是否为空
     */
    public boolean isEmpty() {
        return queue.isEmpty();
    }
    
    /**
     * 获取总生产消息数
     * 
     * @return 总生产消息数
     */
    public long getTotalProduced() {
        return totalProduced.get();
    }
    
    /**
     * 获取总消费消息数
     * 
     * @return 总消费消息数
     */
    public long getTotalConsumed() {
        return totalConsumed.get();
    }
    
    /**
     * 获取待处理消息数
     * 
     * @return 待处理消息数
     */
    public long getPendingCount() {
        return totalProduced.get() - totalConsumed.get();
    }
    
    /**
     * 打印统计信息
     */
    public void printStats() {
        System.out.println("=== 消息队列统计信息 ===");
        System.out.println("总生产消息数: " + totalProduced.get());
        System.out.println("总消费消息数: " + totalConsumed.get());
        System.out.println("待处理消息数: " + getPendingCount());
        System.out.println("当前队列大小: " + size());
        System.out.println("====================");
    }
}

